package com.tjbklx1.redis.chap07;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang3.tuple.Pair;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.SortingParams;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.ZParams;

public class Query {
	public final List<List<String>> all = new ArrayList<List<String>>();
	public final Set<String> unwanted = new HashSet<String>();

	public static final Pattern QUERY_RE = Pattern.compile("[+-]?[a-z']{2,}");
	public static final Pattern WORDS_RE = Pattern.compile("[a-z']{2,}");
	public static final Set<String> STOP_WORDS = new HashSet<String>();
	static {
		for (String word : ("able about across after all almost also am among "
				+ "an and any are as at be because been but by can "
				+ "cannot could dear did do does either else ever "
				+ "every for from get got had has have he her hers " + "him his how however if in into is it its just "
				+ "least let like likely may me might most must my "
				+ "neither no nor not of off often on only or other "
				+ "our own rather said say says she should since so "
				+ "some than that the their them then there these " + "they this tis to too twas us wants was we were "
				+ "what when where which while who whom why will " + "with would yet you your").split(" ")) {
			STOP_WORDS.add(word);
		}
	}

	public static String CONTENT = "this is some random content, look at how it is indexed.";

	public static Set<String> tokenize(String content) {
		Set<String> words = new HashSet<String>();
		Matcher matcher = WORDS_RE.matcher(content);
		while (matcher.find()) {
			String word = matcher.group().trim();
			if (word.length() > 2 && !STOP_WORDS.contains(word)) {
				words.add(word);
			}
		}
		return words;
	}

	public static int indexDocument(Jedis conn, String docid, String content) {
		Set<String> words = tokenize(content);
		Transaction trans = conn.multi();
		for (String word : words) {
			trans.sadd("idx:" + word, docid);
		}
		return trans.exec().size();
	}

	public static Query parse(String queryString) {
		Query query = new Query();
		Set<String> current = new HashSet<String>();
		Matcher matcher = QUERY_RE.matcher(queryString.toLowerCase());
		while (matcher.find()) {
			String word = matcher.group().trim();
			char prefix = word.charAt(0);
			if (prefix == '+' || prefix == '-') {
				word = word.substring(1);
			}

			if (word.length() < 2 || STOP_WORDS.contains(word)) {
				continue;
			}

			if (prefix == '-') {
				query.unwanted.add(word);
				continue;
			}

			if (!current.isEmpty() && prefix != '+') {
				query.all.add(new ArrayList<String>(current));
				current.clear();
			}
			current.add(word);
		}

		if (!current.isEmpty()) {
			query.all.add(new ArrayList<String>(current));
		}
		return query;
	}

	public static String parseAndSearch(Jedis conn, String queryString, int ttl) {
		Query query = parse(queryString);
		if (query.all.isEmpty()) {
			return null;
		}

		List<String> toIntersect = new ArrayList<String>();
		for (List<String> syn : query.all) {
			if (syn.size() > 1) {
				Transaction trans = conn.multi();
				toIntersect.add(union(trans, ttl, syn.toArray(new String[syn.size()])));
				trans.exec();
			} else {
				toIntersect.add(syn.get(0));
			}
		}

		String intersectResult = null;
		if (toIntersect.size() > 1) {
			Transaction trans = conn.multi();
			intersectResult = intersect(trans, ttl, toIntersect.toArray(new String[toIntersect.size()]));
			trans.exec();
		} else {
			intersectResult = toIntersect.get(0);
		}

		if (!query.unwanted.isEmpty()) {
			String[] keys = query.unwanted.toArray(new String[query.unwanted.size() + 1]);
			keys[keys.length - 1] = intersectResult;
			Transaction trans = conn.multi();
			intersectResult = difference(trans, ttl, keys);
			trans.exec();
		}

		return intersectResult;
	}

	private static String setCommon(Transaction trans, String method, int ttl, String... items) {
		String[] keys = new String[items.length];
		for (int i = 0; i < items.length; i++) {
			keys[i] = "idx:" + items[i];
		}

		String id = UUID.randomUUID().toString();
		try {
			trans.getClass().getDeclaredMethod(method, String.class, String[].class).invoke(trans, "idx:" + id, keys);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		trans.expire("idx:" + id, ttl);
		return id;
	}

	public static String intersect(Transaction trans, int ttl, String... items) {
		return setCommon(trans, "sinterstore", ttl, items);
	}

	public static String union(Transaction trans, int ttl, String... items) {
		return setCommon(trans, "sunionstore", ttl, items);
	}

	public static String difference(Transaction trans, int ttl, String... items) {
		return setCommon(trans, "sdiffstore", ttl, items);
	}

	@SuppressWarnings("unchecked")
	public static SearchResult searchAndSort(Jedis conn, String queryString, String sort) {
		boolean desc = sort.startsWith("-");
		if (desc) {
			sort = sort.substring(1);
		}
		boolean alpha = !"updated".equals(sort) && !"id".equals(sort);
		String by = "kb:doc:*->" + sort;

		String id = parseAndSearch(conn, queryString, 300);

		Transaction trans = conn.multi();
		trans.scard("idx:" + id);
		SortingParams params = new SortingParams();
		if (desc) {
			params.desc();
		}
		if (alpha) {
			params.alpha();
		}
		params.by(by);
		params.limit(0, 20);
		trans.sort("idx:" + id, params);
		List<Object> results = trans.exec();

		return new SearchResult(id, ((Long) results.get(0)).longValue(), (List<String>) results.get(1));
	}

	private static String zsetCommon(Transaction trans, String method, int ttl, ZParams params, String... sets) {
		String[] keys = new String[sets.length];
		for (int i = 0; i < sets.length; i++) {
			keys[i] = "idx:" + sets[i];
		}

		String id = UUID.randomUUID().toString();
		try {
			trans.getClass().getDeclaredMethod(method, String.class, ZParams.class, String[].class)
					.invoke(trans, "idx:" + id, params, keys);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		trans.expire("idx:" + id, ttl);
		return id;
	}

	public static String zintersect(Transaction trans, int ttl, ZParams params, String... sets) {
		return zsetCommon(trans, "zinterstore", ttl, params, sets);
	}

	public static String zunion(Transaction trans, int ttl, ZParams params, String... sets) {
		return zsetCommon(trans, "zunionstore", ttl, params, sets);
	}

	@SuppressWarnings("unchecked")
	public static SearchResult searchAndZsort(Jedis conn, String queryString, boolean desc, Map<String, Integer> weights) {
		int ttl = 300;
		int start = 0;
		int num = 20;
		String id = parseAndSearch(conn, queryString, ttl);

		int updateWeight = weights.containsKey("update") ? weights.get("update") : 1;
		int voteWeight = weights.containsKey("vote") ? weights.get("vote") : 0;

		String[] keys = new String[] { id, "sort:update", "sort:votes" };
		Transaction trans = conn.multi();
		id = zintersect(trans, ttl, new ZParams().weights(0, updateWeight, voteWeight), keys);

		trans.zcard("idx:" + id);
		if (desc) {
			trans.zrevrange("idx:" + id, start, start + num - 1);
		} else {
			trans.zrange("idx:" + id, start, start + num - 1);
		}
		List<Object> results = trans.exec();

		return new SearchResult(id, ((Long) results.get(results.size() - 2)).longValue(),
		// Note: it's a LinkedHashSet, so it's ordered
				new ArrayList<String>((Set<String>) results.get(results.size() - 1)));
	}

	public static long stringToScore(String string) {
		return stringToScore(string, false);
	}

	public static long stringToScore(String string, boolean ignoreCase) {
		if (ignoreCase) {
			string = string.toLowerCase();
		}

		List<Integer> pieces = new ArrayList<Integer>();
		for (int i = 0; i < Math.min(string.length(), 6); i++) {
			pieces.add((int) string.charAt(i));
		}
		while (pieces.size() < 6) {
			pieces.add(-1);
		}

		long score = 0;
		for (int piece : pieces) {
			score = score * 257 + piece + 1;
		}

		return score * 2 + (string.length() > 6 ? 1 : 0);
	}

	public static long stringToScoreGeneric(String string, Map<Integer, Integer> mapping) {
		int length = (int) (52 / (Math.log(mapping.size()) / Math.log(2)));

		List<Integer> pieces = new ArrayList<Integer>();
		for (int i = 0; i < Math.min(string.length(), length); i++) {
			pieces.add((int) string.charAt(i));
		}
		while (pieces.size() < 6) {
			pieces.add(-1);
		}

		long score = 0;
		for (int piece : pieces) {
			int value = mapping.get(piece);
			score = score * mapping.size() + value + 1;
		}

		return score * 2 + (string.length() > 6 ? 1 : 0);
	}

	public static long zaddString(Jedis conn, String name, Map<String, String> values) {
		Map<String,Double> pieces = new HashMap<String,Double>(values.size());
		for (Map.Entry<String, String> entry : values.entrySet()) {
			pieces.put(entry.getKey(),(double) stringToScore(entry.getValue()));
		}
		conn.zadd(name, pieces);
		return 1L;
	}

	private static Map<Ecpm, Double> AVERAGE_PER_1K = new HashMap<Ecpm, Double>();

	public static void indexAd(Jedis conn, String id, String[] locations, String content, Ecpm type, double value) {
		Transaction trans = conn.multi();

		for (String location : locations) {
			trans.sadd("idx:req:" + location, id);
		}

		Set<String> words = tokenize(content);
		for (String word : tokenize(content)) {
			trans.zadd("idx:" + word, 0, id);
		}

		double avg = AVERAGE_PER_1K.containsKey(type) ? AVERAGE_PER_1K.get(type) : 1;
		double rvalue = toEcpm(type, 1000, avg, value);

		trans.hset("type:", id, type.name().toLowerCase());
		trans.zadd("idx:ad:value:", rvalue, id);
		trans.zadd("ad:base_value:", value, id);
		for (String word : words) {
			trans.sadd("terms:" + id, word);
		}
		trans.exec();
	}

	public static double toEcpm(Ecpm type, double views, double avg, double value) {
		switch (type) {
		case CPC:
		case CPA:
			return 1000. * value * avg / views;
		case CPM:
			return value;
		}
		return value;
	}

	@SuppressWarnings("unchecked")
	public static Pair<Long, String> targetAds(Jedis conn, String[] locations, String content) {
		Transaction trans = conn.multi();

		String matchedAds = matchLocation(trans, locations);

		String baseEcpm = zintersect(trans, 30, new ZParams().weights(0, 1), matchedAds, "ad:value:");

		Pair<Set<String>, String> result = finishScoring(trans, matchedAds, baseEcpm, content);

		trans.incr("ads:served:");
		trans.zrevrange("idx:" + result.getValue(), 0, 0);

		List<Object> response = trans.exec();
		long targetId = (Long) response.get(response.size() - 2);
		Set<String> targetedAds = (Set<String>) response.get(response.size() - 1);

		if (targetedAds.size() == 0) {
			return Pair.of(null, null);
		}

		String adId = targetedAds.iterator().next();
		recordTargetingResult(conn, targetId, adId, result.getKey());

		return Pair.of(targetId, adId);
	}

	public static String matchLocation(Transaction trans, String[] locations) {
		String[] required = new String[locations.length];
		for (int i = 0; i < locations.length; i++) {
			required[i] = "req:" + locations[i];
		}
		return union(trans, 300, required);
	}

	public static Pair<Set<String>, String> finishScoring(Transaction trans, String matched, String base, String content) {
		Map<String, Integer> bonusEcpm = new HashMap<String, Integer>();
		Set<String> words = tokenize(content);
		for (String word : words) {
			String wordBonus = zintersect(trans, 30, new ZParams().weights(0, 1), matched, word);
			bonusEcpm.put(wordBonus, 1);
		}

		if (bonusEcpm.size() > 0) {

			String[] keys = new String[bonusEcpm.size()];
			int[] weights = new int[bonusEcpm.size()];
			int index = 0;
			for (Map.Entry<String, Integer> bonus : bonusEcpm.entrySet()) {
				keys[index] = bonus.getKey();
				weights[index] = bonus.getValue();
				index++;
			}

			ZParams minParams = new ZParams().aggregate(ZParams.Aggregate.MIN).weights(weights);
			String minimum = zunion(trans, 30, minParams, keys);

			ZParams maxParams = new ZParams().aggregate(ZParams.Aggregate.MAX).weights(weights);
			String maximum = zunion(trans, 30, maxParams, keys);

			String result = zunion(trans, 30, new ZParams().weights(2, 1, 1), base, minimum, maximum);
			return Pair.of(words, result);
		}
		return Pair.of(words, base);
	}

	public static void recordTargetingResult(Jedis conn, long targetId, String adId, Set<String> words) {
		Set<String> terms = conn.smembers("terms:" + adId);
		String type = conn.hget("type:", adId);

		Transaction trans = conn.multi();
		terms.addAll(words);
		if (terms.size() > 0) {
			String matchedKey = "terms:matched:" + targetId;
			for (String term : terms) {
				trans.sadd(matchedKey, term);
			}
			trans.expire(matchedKey, 900);
		}

		trans.incr("type:" + type + ":views:");
		for (String term : terms) {
			trans.zincrby("views:" + adId, 1, term);
		}
		trans.zincrby("views:" + adId, 1, "");

		List<Object> response = trans.exec();
		double views = (Double) response.get(response.size() - 1);
		if ((views % 100) == 0) {
			updateCpms(conn, adId);
		}
	}

	@SuppressWarnings("unchecked")
	public static void updateCpms(Jedis conn, String adId) {
		Transaction trans = conn.multi();
		trans.hget("type:", adId);
		trans.zscore("ad:base_value:", adId);
		trans.smembers("terms:" + adId);
		List<Object> response = trans.exec();
		String type = (String) response.get(0);
		Double baseValue = (Double) response.get(1);
		Set<String> words = (Set<String>) response.get(2);

		String which = "clicks";
		Ecpm ecpm = Enum.valueOf(Ecpm.class, type.toUpperCase());
		if (Ecpm.CPA.equals(ecpm)) {
			which = "actions";
		}

		trans = conn.multi();
		trans.get("type:" + type + ":views:");
		trans.get("type:" + type + ':' + which);
		response = trans.exec();
		String typeViews = (String) response.get(0);
		String typeClicks = (String) response.get(1);

		AVERAGE_PER_1K.put(
				ecpm,
				1000. * Integer.valueOf(typeClicks != null ? typeClicks : "1")
						/ Integer.valueOf(typeViews != null ? typeViews : "1"));

		if (Ecpm.CPM.equals(ecpm)) {
			return;
		}

		String viewKey = "views:" + adId;
		String clickKey = which + ':' + adId;

		trans = conn.multi();
		trans.zscore(viewKey, "");
		trans.zscore(clickKey, "");
		response = trans.exec();
		Double adViews = (Double) response.get(0);
		Double adClicks = (Double) response.get(1);

		double adEcpm = 0;
		if (adClicks == null || adClicks < 1) {
			Double score = conn.zscore("idx:ad:value:", adId);
			adEcpm = score != null ? score.doubleValue() : 0;
		} else {
			adEcpm = toEcpm(ecpm, adViews != null ? adViews.doubleValue() : 1,
					adClicks != null ? adClicks.doubleValue() : 0, baseValue);
			conn.zadd("idx:ad:value:", adEcpm, adId);
		}
		for (String word : words) {
			trans = conn.multi();
			trans.zscore(viewKey, word);
			trans.zscore(clickKey, word);
			response = trans.exec();
			Double views = (Double) response.get(0);
			Double clicks = (Double) response.get(1);

			if (clicks == null || clicks < 1) {
				continue;
			}

			double wordEcpm = toEcpm(ecpm, views != null ? views.doubleValue() : 1,
					clicks != null ? clicks.doubleValue() : 0, baseValue);
			double bonus = wordEcpm - adEcpm;
			conn.zadd("idx:" + word, bonus, adId);
		}
	}

	public static void recordClick(Jedis conn, long targetId, String adId, boolean action) {
		String type = conn.hget("type:", adId);
		Ecpm ecpm = Enum.valueOf(Ecpm.class, type.toUpperCase());

		String clickKey = "clicks:" + adId;
		String matchKey = "terms:matched:" + targetId;
		Set<String> matched = conn.smembers(matchKey);
		matched.add("");

		Transaction trans = conn.multi();
		if (Ecpm.CPA.equals(ecpm)) {
			trans.expire(matchKey, 900);
			if (action) {
				clickKey = "actions:" + adId;
			}
		}

		if (action && Ecpm.CPA.equals(ecpm)) {
			trans.incr("type:" + type + ":actions:");
		} else {
			trans.incr("type:" + type + ":clicks:");
		}

		for (String word : matched) {
			trans.zincrby(clickKey, 1, word);
		}
		trans.exec();

		updateCpms(conn, adId);
	}

	public static void addJob(Jedis conn, String jobId, String... requiredSkills) {
		conn.sadd("job:" + jobId, requiredSkills);
	}

	@SuppressWarnings("unchecked")
	public static boolean isQualified(Jedis conn, String jobId, String... candidateSkills) {
		String temp = UUID.randomUUID().toString();
		Transaction trans = conn.multi();
		for (String skill : candidateSkills) {
			trans.sadd(temp, skill);
		}
		trans.expire(temp, 5);
		trans.sdiff("job:" + jobId, temp);

		List<Object> response = trans.exec();
		Set<String> diff = (Set<String>) response.get(response.size() - 1);
		return diff.size() == 0;
	}

	public static void indexJob(Jedis conn, String jobId, String... skills) {
		Transaction trans = conn.multi();
		Set<String> unique = new HashSet<String>();
		for (String skill : skills) {
			trans.sadd("idx:skill:" + skill, jobId);
			unique.add(skill);
		}
		trans.zadd("idx:jobs:req", unique.size(), jobId);
		trans.exec();
	}

	public static Set<String> findJobs(Jedis conn, String... candidateSkills) {
		String[] keys = new String[candidateSkills.length];
		int[] weights = new int[candidateSkills.length];
		for (int i = 0; i < candidateSkills.length; i++) {
			keys[i] = "skill:" + candidateSkills[i];
			weights[i] = 1;
		}

		Transaction trans = conn.multi();
		String jobScores = zunion(trans, 30, new ZParams().weights(weights), keys);
		String finalResult = zintersect(trans, 30, new ZParams().weights(-1, 1), jobScores, "jobs:req");
		trans.exec();

		return conn.zrangeByScore("idx:" + finalResult, 0, 0);
	}

}